//
// Created by 郝羽 on 2018/5/11.
//

#include "proxy.hpp"

#include <libsoup/soup.h>
#include <map>
#include <string>

#include "xm_app_config.h"
#include "xm_log.h"
#include "rapidjson/document.h"
#include "rapidjson/prettywriter.h"
#include "rapidjson/reader.h"
#include "xcutil.h"
#include "xrtc_endpoint.h"

// libsoup 例子
// https://github.com/onurmark/soup-example/blob/83dc6a3d4b4dd9ed69f2de7dcff5fb539caea3f0/src/soup_example.c

constexpr int kConnectTimeout = 3000;
constexpr int kRegisterTimeout = 1000;
constexpr int kKeepaliveTimeout = 30000;

struct proxy_rpc {
    std::string op;
    int64_t expire_time;
};

struct proxy {
    GMainContext* ctx;
    std::string proxy_addr;
    SoupSession* session;
    SoupWebsocketConnection* client;
    bool connected;
    bool registered;
    int register_id;
    int rpc_id;
    GSource* timeout;
    GCancellable* cancel_connect;
    int64_t last_connect_time_ms;
    int64_t last_register_time_ms;
    int64_t last_keepalive_time_ms;
    std::map<std::string, std::string> conf_id_by_rtc_id;
    std::map<int, proxy_rpc> rpc_map;
};

void proxy_connect(proxy* p);
void proxy_connect_finish(GObject *obj, GAsyncResult *result, gpointer data);
gboolean proxy_timeout_handler(gpointer data);
void proxy_error(SoupWebsocketConnection* ws, GError *error, gpointer user_data);
void proxy_closed(SoupWebsocketConnection* ws, gpointer data);
void proxy_register(proxy* p);
void proxy_message(SoupWebsocketConnection* ws, SoupWebsocketDataType dt, GBytes* msg, gpointer data);
void proxy_keepalive(proxy* p);

proxy* proxy_new(GMainContext* ctx, const std::string& proxyAddr, const std::string& httpAddr) {
    auto p = new proxy();
    p->ctx = ctx;
    p->proxy_addr = proxyAddr;
    p->session = soup_session_new();
    p->client = nullptr;
    p->register_id = -1;

    p->timeout = g_timeout_source_new(1000);

    g_source_set_callback(p->timeout, proxy_timeout_handler, p, nullptr);
    g_source_attach(p->timeout, p->ctx);

    proxy_connect(p);
    return p;
}

void proxy_delete(proxy* p) {
    g_source_remove(g_source_get_id(p->timeout));
    g_source_unref(p->timeout);
    if (p->cancel_connect) {
        g_cancellable_cancel(p->cancel_connect);
        g_object_unref(p->cancel_connect);
        p->cancel_connect = nullptr;
    }
    delete p;
}

void proxy_connect(proxy* p) {
    auto url = g_strdup_printf("ws://%s/proxy", p->proxy_addr.c_str());
    dbgi("connecting to %s", url);
    auto get = soup_message_new("GET", url);
    g_free(url);

    p->cancel_connect = g_cancellable_new();
    soup_session_websocket_connect_async(p->session, get, nullptr, nullptr, p->cancel_connect, proxy_connect_finish, p);
}

gboolean proxy_timeout_handler(gpointer data) {
    auto now_ms = get_timestamp_ms();
    auto p = static_cast<proxy*>(data);
    if (!p->connected) {
        auto elapsed_ms = now_ms - p->last_connect_time_ms;
        if (elapsed_ms >= kConnectTimeout) {
            if (p->cancel_connect) {
                g_cancellable_cancel(p->cancel_connect);
                g_object_unref(p->cancel_connect);
                p->cancel_connect = nullptr;
            }
            proxy_connect(p);
            p->last_connect_time_ms = now_ms;
        }
    } else if (!p->registered) {
        auto elapsed_ms = now_ms - p->last_register_time_ms;
        if (elapsed_ms >= kRegisterTimeout) {
            proxy_register(p);
            p->last_register_time_ms = now_ms;
        }
    } else {
        auto elapsed_ms = now_ms - p->last_keepalive_time_ms;
        if (elapsed_ms > kKeepaliveTimeout) {
            proxy_keepalive(p);
            p->last_keepalive_time_ms = now_ms;
        }
    }

    auto rpc_it = p->rpc_map.begin();
    while (rpc_it != p->rpc_map.end()) {
        auto rpc_it_cur = rpc_it++;
        auto rpc = rpc_it_cur->second;
        if (rpc.expire_time < now_ms) {
            p->rpc_map.erase(rpc_it_cur);
        }
    }

    return true;
}

void proxy_connect_finish(GObject *obj, GAsyncResult *result, gpointer data) {
    auto p = static_cast<proxy*>(data);
    GError* err = nullptr;
    p->client = soup_session_websocket_connect_finish(SOUP_SESSION(obj), result, &err);
    if (err != nullptr) {
        dbge("proxy: connect error: %s", err->message);
        g_error_free(err);
        return;
    }

    GValue keepalive = G_VALUE_INIT;
    g_value_init(&keepalive, G_TYPE_UINT);
    g_value_set_uint(&keepalive, 3);
    g_object_set_property(G_OBJECT(p->client), "keepalive-interval", &keepalive);
    g_value_unset(&keepalive);

    g_signal_connect(p->client, "error", G_CALLBACK(proxy_error), p);
    g_signal_connect(p->client, "closed", G_CALLBACK(proxy_closed), p);
    g_signal_connect(p->client, "message", G_CALLBACK(proxy_message), p);

    p->connected = true;
    p->registered = false;

    proxy_register(p);
}

void proxy_error(SoupWebsocketConnection* ws, GError *error, gpointer user_data) {
    dbge("proxy: error %s", error->message);
}

void proxy_closed(SoupWebsocketConnection* ws, gpointer data) {
    auto p = static_cast<proxy*>(data);
    p->connected = false;
    p->registered = false;
    dbgi("proxy: disconnected");
}

void proxy_register(proxy* p) {
    auto rpc_id = ++p->rpc_id;

    p->rpc_map.emplace(rpc_id, proxy_rpc{.op = "register", .expire_time = get_timestamp_ms() + kRegisterTimeout});

    auto config = app_config_get();

    auto http_addr = g_strdup_printf("http://%s:%d/rtc/", config->http_ip, config->http_port);

    rapidjson::StringBuffer sb;
    //rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(sb);
    rapidjson::Writer<rapidjson::StringBuffer> writer(sb);

    writer.StartObject();
    writer.Key("op");
    writer.String("register");
    writer.Key("url");
    writer.String(http_addr);
    writer.Key("backend-id");
    writer.Int(p->register_id);
    writer.Key("rpc-id");
    writer.Int(rpc_id);
    writer.EndObject();

//    auto reg = g_strdup_printf("{\"op\":\"register\",\"url\":\"%s\",\"backendId\":%d,\"rpcId\":%d}",
//                               http_addr, p->registerId, p->registerRpcId);

    auto reg = sb.GetString();

    if (p->client) {
        dbgi("proxy send register: %s", reg);
        soup_websocket_connection_send_text(p->client, reg);
    } else {
        dbgi("proxy could not send register: not connected");
    }

    // g_free(reg);
    g_free(http_addr);
}

void proxy_message(SoupWebsocketConnection* ws, SoupWebsocketDataType dt, GBytes* msg, gpointer data) {
    auto p = static_cast<proxy*>(data);
    if (dt == SOUP_WEBSOCKET_DATA_TEXT) {
        gsize len;
        auto text = static_cast<const char*>(g_bytes_get_data(msg, &len));

        dbgi("received text message: %s", text);

        rapidjson::Document doc;
        doc.Parse(text);

        auto it_op = doc.FindMember("op");
        if (it_op == doc.MemberEnd()) {
            dbgw("received invalid msg: 'op' not found");
            return;
        }

        if (!it_op->value.IsString()) {
            dbgw("received invalid msg: 'op' not string");
            return;
        }

        std::string op(it_op->value.GetString(), it_op->value.GetStringLength());
        if (op != "response") {
            dbgw("received invalid msg: 'op' %s", op.c_str());
            return;
        }

        auto it_status = doc.FindMember("status");
        if (it_status == doc.MemberEnd()) {
            dbgw("received invalid msg: 'status' not found");
            return;
        }

        if (!it_status->value.IsInt()) {
            dbgw("received invalid msg: 'status' not int");
            return;
        }

        auto status = it_status->value.GetInt();

        auto it_rpcid = doc.FindMember("rpc-id");
        if (it_rpcid == doc.MemberEnd()) {
            dbgw("received invalid msg: 'rpc-id' not found");
            return;
        }

        if (!it_rpcid->value.IsInt()) {
            dbgw("received invalid msg: 'rpc-id' not int");
            return;
        }

        auto rpc_id = it_rpcid->value.GetInt();
        auto rpc_it = p->rpc_map.find(rpc_id);

        if (rpc_it == p->rpc_map.end()) {
            dbgi("received invalid rpc-id: %d", rpc_id);
            return;
        }

        auto& rpc = rpc_it->second;

        if (strcmp(rpc.op.c_str(), "register") == 0) {
            auto it_backend_id = doc.FindMember("backend-id");

            if (it_backend_id != doc.MemberEnd() && it_backend_id->value.IsInt()) {
                auto backend_id = it_backend_id->value.GetInt();
                p->register_id = backend_id;
                dbgi("received register response backend-id: %d", backend_id);
            }
            dbgi("received register response status: %d", status);
            p->registered = true;
        }

        p->rpc_map.erase(rpc_it);
    } else if (dt == SOUP_WEBSOCKET_DATA_BINARY) {
        dbgi("received binary message");
    }
}

void proxy_keepalive(proxy* p) {
    if (!p->registered) {
        dbgi("could not send keepalive: not registered");
        return;
    }
    std::map<std::string, std::list<std::string>> rtc_id_list_by_conf_id;
    for (auto it : p->conf_id_by_rtc_id) {
        auto rtc_id = it.first;
        auto conf_id = it.second;
        rtc_id_list_by_conf_id[conf_id].emplace_back(rtc_id);
    }

    rapidjson::StringBuffer sb;
    //rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(sb);
    rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
    writer.StartObject();
    writer.Key("op");
    writer.String("keepalive");
    writer.Key("conf-list");
    writer.StartArray();
    for (auto it : rtc_id_list_by_conf_id) {
        auto conf_id = it.first;
        auto rtc_id_list = it.second;
        writer.StartObject();
        writer.Key("id");
        writer.String(conf_id);
        writer.Key("rtc-list");
        writer.StartArray();
        for (auto rtc_id : rtc_id_list) {
            writer.String(rtc_id);
        }
        writer.EndArray();
        writer.EndObject();
    }
    writer.EndArray();
    writer.EndObject();

    if (p->client) {
        dbgd("proxy send keepalive: %s", sb.GetString());
        soup_websocket_connection_send_text(p->client, sb.GetString());
    } else {
        dbgw("proxy could not send keepalive: not connected");
    }
}

void proxy_add_keepalive_rtc(proxy* p, const std::string& conf_id, const std::string& rtc_id) {
    dbgi("proxy: add conf '%s' rtc '%s' to keep alive list", conf_id.c_str(), rtc_id.c_str());
    p->conf_id_by_rtc_id.emplace(rtc_id, conf_id);
}

void proxy_del_keepalive_rtc(proxy* p, const std::string& conf_id, const std::string& rtc_id) {

    dbgi("proxy: del rtc %s", rtc_id.c_str());
    p->conf_id_by_rtc_id.erase(rtc_id);

    rapidjson::StringBuffer sb;
    //rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(sb);
    rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
    writer.StartObject();
    writer.Key("op");
    writer.String("rtc-destroy");
    writer.Key("rtc-id");
    writer.String(rtc_id);
    writer.Key("conf-id");
    writer.String(conf_id);

    writer.EndObject();

    if (p->client) {
        dbgi("proxy send rtc-destroy: %s", sb.GetString());
        soup_websocket_connection_send_text(p->client, sb.GetString());
    } else {
        dbgi("proxy could not send rtc-destroy: not connected %s", rtc_id.c_str());
    }
    dbgi("proxy: success del rtc %s", rtc_id.c_str());
}
